package com.flink.demo.redis;

import com.kingstar.flink.table.AbstractDataSource;
import com.kingstar.flink.utils.RowTypeUtils;
import org.apache.commons.lang3.RandomUtils;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.types.Row;

/**
 * 自定义row数据源
 *
 * @author xin.ding
 * @date 2021/3/18 14:13
 */
public class RowDataSource extends AbstractDataSource<Row> {

    /**
     * 生成数据条数
     */
    private int num = 10;

    @Override
    public TypeInformation setResultType() {
        return RowTypeUtils.buildRowTypeInfo(new String[][]{{"id", "string"}});
    }

    @Override
    public void run(SourceContext<Row> out) throws Exception {
        for (int i = 0; i < num; i++) {
            String value = String.format("%d", RandomUtils.nextInt(1, 4));
            Row row = new Row(1);
            row.setField(0, value);
            out.collect(row);
        }
    }

    @Override
    public void cancel() {

    }
}
